/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.values.IntType;
import edu.ucsd.db.bassarisk.data.values.IntType.IntTools;
import edu.ucsd.db.bassarisk.data.values.IntType.IntegerValueCodec;

/**
 * Used to read a sequence of 32-bit integers from a column in the Bassarisk
 * column store database.
 * 
 * This class reads a single Card with a simple fixed-width integer
 * serialization.
 * 
 * @author nbales
 */
public class FixedLengthIntegerCardReader extends AbstractCardReader<IntType> {

  /** The size of a serialized integer in a fixed width scheme */
  private final static int INTEGER_SIZE = 4;

  /**
   * An integer posting reference instance that can interpret generic
   * IPostingReference for sending posting values to callbacks.
   */
  private final IntegerValueCodec integerPosting;

  /**
   * Create a new FixedIntegerCardLoader and initialize final fields.
   */
  public FixedLengthIntegerCardReader() {
    super(IntTools.INSTANCE);
    this.integerPosting = IntTools.INSTANCE.getGenericCodec();
  }

  // Javadoc in IColumnPartReader.java
  @Override
  public void jumpCursor(final int targetEntryIndex,
                         final IValueReference<IntType> valueOutput) {
    // Check condition that request must be on this card
    assert targetEntryIndex >= this.cardInfo.getFirstEntryIndex();
    assert targetEntryIndex < this.cardInfo.getFirstEntryIndex()
        + this.cardInfo.getEntryCount();

    long jumpDistance = targetEntryIndex - this.current.index;

    // Cards should only be read in a forward only manner
    assert jumpDistance >= 0;

    this.current.index = targetEntryIndex;
    this.current.byteOffset += INTEGER_SIZE * jumpDistance;

    this.stepCursor(valueOutput);
  }

  // Javadoc in IColumnPartReader.java
  @Override
  public boolean
      jumpCursorToValue(final IValueReference<IntType> minimumTargetValue,
                        final IValueReference<IntType> valueOut) {
    int targetValue = this.integerPosting.getIntegerValue(minimumTargetValue);

    // TODO (nbales) - Make this faster using a binary search
    int pivot = this.current.byteOffset;
    while (this.stepCursor(valueOut)
        && this.drawerBuffer.getInt(pivot) < targetValue) {
      pivot += INTEGER_SIZE;
    }
    this.current.byteOffset = pivot;

    return this.drawerBuffer.getInt(pivot) >= targetValue;
  }

  // Javadoc in IColumnPartReader.java
  @Override
  public boolean stepCursor(final IValueReference<IntType> valueOut) { 
    if (this.cardInfo == null) { return false; } 
    long maxIndex =
        this.cardInfo.getFirstEntryIndex() + this.cardInfo.getEntryCount();
    if (this.current.index >= maxIndex) { return false; }

    // Sanity check. The full length of the read should be on this card.
    assert this.sanityCheckByte(this.current.byteOffset);
    assert this.sanityCheckByte(this.current.byteOffset + INTEGER_SIZE - 1);

    // TODO (nbales) - Replace with more efficient implementation
    int value = this.drawerBuffer.getInt(this.current.byteOffset);
    this.integerPosting.setIntegerValue(value, valueOut);

    // Advance cursor to next posting
    this.current.index++;
    this.current.byteOffset += INTEGER_SIZE; 
    return true;
  }
}
